/**
 Created by:
 muhtearjan mehmood مۇختەرجان مەخمۇت (Email: yeganaaa@163.com)
 At 6/17/20 10:47 PM
 */
import {RpcBridge, IRpcBridgeHost, IRpcServerHost, StateModuleInMemoryImplementation, IStateModule, BridgeHostEvents, Fragment, ServerHostEvents, Completer, KeepAliveRpcBridgeHost} from "dream_rpc";
import { Channels, IMessageBusClientInterface, IMessageBusServerInterface, VirtualChannel } from "./Interfaces";
import {IClosingReason} from "dream_rpc/bin/Core/RpcBridge";
enum HandlerKind {
    Server = 0x01,
    Bridge = 0x02,
}
type DataReceivedHandler = {
    handler: BridgeHostEvents | ServerHostEvents
    kind: HandlerKind
}
class MessageBusClientImplementation implements IMessageBusClientInterface {
    stateModule: IStateModule = new StateModuleInMemoryImplementation()
    public handlers = new Map<string, DataReceivedHandler>()
    public textDecoder = new TextDecoder()
    public textEncoder = new TextEncoder()
    public constructor(private _owner: MessageBusConnection) {

    }
    public async onChannelConnected(channels: Channels): Promise<void> {
        const handler = this.handlers.get(channels.local.name)
        if (handler == null) {
            const errorMessage = `The channel ${channels.local.name} is not found on this connection`
            console.error(errorMessage)
            throw new Error(errorMessage)
        }
        if (handler.kind == HandlerKind.Bridge) {
            const errorMessage = `The handler kind ${channels.local.name} is not server mode.`
            console.error(errorMessage)
            throw new Error(errorMessage)
        }
        const clientHost = new MessageBusRpcBridgeHost(this._owner, this, { local: channels.remote, remote: channels.local });
        clientHost.createdByServer = true;
        (handler.handler as ServerHostEvents).onClientConnected(clientHost)
    }
    public async onChannelClosed(channels: Channels): Promise<void> {
        let handler = this.handlers.get(channels.local.name)!
        function _checkHandler() {
            if (handler == null) {
                const errorMessage = `Event handler of channel ${channels.local.name} is not found`
                console.error(errorMessage)
                throw new Error(errorMessage)
            }
        }
        _checkHandler()
        if (handler.kind == HandlerKind.Bridge) await (handler.handler as BridgeHostEvents).onClose()
        if (handler.kind == HandlerKind.Server) {
            handler = this.handlers.get(channels.remote.name)!
            _checkHandler()
            await (handler.handler as BridgeHostEvents).onClose()
        }
    }
    public async onDataReceived(channelWithData: Uint8Array): Promise<void> {
        const fragments = [...Fragment.fromBytes(channelWithData)]
        const channels = JSON.parse(this.textDecoder.decode(fragments[0].data)) as Channels
        let handler = this.handlers.get(channels.local.name)
        if (handler == null) {
            const errorMessage = `the local virtual channel ${channels.local.name} is not exists, but it is still receive some message...`
            console.error(errorMessage)
            throw new Error(errorMessage)
        }
        if (handler.kind != HandlerKind.Bridge) {
            handler = this.handlers.get(channels.remote.name)
            if (handler == null) {
                const errorMessage = `either handler kind is should be Bridge and not Server or a handler that kind is Bridge is should be exists, but the handler kind ${channels.local.name} is Server mode and a handler that kind type is Bridge is not exists...`
                console.error(errorMessage)
                throw new Error(errorMessage)
            }
        }
        (handler.handler as BridgeHostEvents).onMessage(fragments[1].data)
    }
    public async keepAlive(): Promise<void> {

    }
    public async sendErrorMessage(error: string): Promise<void> {
        console.error(error)
    }
}

export class MessageBusRpcBridgeHost implements IRpcBridgeHost<MessageBusClientImplementation> {
    private _eventHandler: BridgeHostEvents = null!
    createdByServer: boolean = false
    private _waitForOpen = Completer.create()
    private _hasConnected = false
    public constructor(private _owner: MessageBusConnection, public backend: MessageBusClientImplementation, private channels: Channels) {

    }
    public async prepare(eventHandler: BridgeHostEvents): Promise<void> {
        this._eventHandler = eventHandler
        this.backend.handlers.set(this.channels.local.name, {
            kind: HandlerKind.Bridge,
            handler: new BridgeHostEvents(
                async () => { },
                async error => { },
                async () => {
                    await this._waitForOpen.promise
                    this.backend.handlers.delete(this.channels.local.name)
                    await this._dispatchClosedEvent()
                },
                async () => {},
                async data => {
                    await this._waitForOpen.promise
                    await this._eventHandler.onMessage(data)
                },
            )
        });
        (async () => {
            await this._waitForOpen.promise
            await this._eventHandler.onReady()
            this._hasConnected = true
        })().catch(reason => {})
    }
    public async open(): Promise<void> {
        try {
            if (!this.createdByServer) await this._owner.bridge.remoteInterface.connect(this.channels);
            this._waitForOpen.complete()
        } catch (error) {
            this._waitForOpen.error(error)
            await this._eventHandler.onError(error)
            await this._eventHandler.onUnableToConnect()
        }
    }
    public async close(reason: IClosingReason): Promise<void> {
        try {
            await this._owner.bridge.remoteInterface.close(this.channels.local)
        }
        catch (error) {}
        this.backend.handlers.delete(this.channels.local.name)
        await this._dispatchClosedEvent()
    }
    private async _dispatchClosedEvent() {
        if (!this._hasConnected) return
        this._hasConnected = false
        await this._eventHandler.onClose()
    }
    public async send(data: Uint8Array): Promise<void> {
        let channels: Channels = this.channels
        if (this.createdByServer) {
            channels = {
                local: this.channels.remote,
                remote: channels.local,
            }
        }
        const channelsWithData = Fragment.encodeAll(
            new Fragment(this.backend.textEncoder.encode(JSON.stringify(channels))),
            new Fragment(data)
        )
        await this._owner.bridge.remoteInterface.sendData(channelsWithData)
    }
}
export class MessageBusRpcServerHost implements IRpcServerHost<MessageBusClientImplementation> {
    private _eventHandler: ServerHostEvents = null!
    private _waitForLaunch = Completer.create()

    public constructor(public backend: MessageBusClientImplementation, private _owner: MessageBusConnection, private _channel: VirtualChannel) { }
    public async prepare(eventHandler: ServerHostEvents): Promise<void> {
        this._eventHandler = eventHandler
        this.backend.handlers.set(this._channel.name, {
            kind: HandlerKind.Server,
            handler: new ServerHostEvents(
                async () => { },
                async bridgeOfClient => {
                    await this._eventHandler.onClientConnected(bridgeOfClient)
                },
                async error => { },
                async () => {
                    this.backend.handlers.delete(this._channel.name)
                    await this._eventHandler.onShutdown()
                }
            )
        })
        await this._owner.bridge.remoteInterface.create(this._channel)
        this._eventHandler.onReady()
    }
    public async launch(): Promise<void> {
        this._waitForLaunch.complete()
    }
    public async shutdown(): Promise<void> {
        try {
            await this._owner.bridge.remoteInterface.close(this._channel)
        }
        catch (e) {}
        this.backend.handlers.delete(this._channel.name)
        this._eventHandler.onShutdown()
    }
}

/**
 * Message bus connection class
 * @remarks
 * This class represents a message bus server client, through this, you can create more {@link IRpcBridgeHost} and {@link IRpcServerHost} instances over this connection.
 * this is very useful when you want to create more RpcClient/RpcServer over single tcp/web socket (or other) connection.
 * on some environments impossible to create Socket server so on these environment no way to create rpc server host
 * but using this, you can using {@link createServerHost} method to create {@link IRpcServerHost}
 * and
 * using {@link createBridgeHost} method to create {@link IRpcBridgeHost} instance.
 */
export class MessageBusConnection {
    /**
     * Do not use this if not needed, this is used only by internal implementation over this RpcBridge
     */
    public bridge: RpcBridge<IMessageBusClientInterface, IMessageBusServerInterface> = null!
    /**
     * used to increase internally allocated id for server host.
     * @private
     */
    private _counter = 0
    private _hosts = new Map<number, IntermediateServerHost>()
    private _allocateId(): number {
        return this._counter++
    }
    /**
     * Creator
     * @param bridgeHostCreator Some bridge host
     * @param bridgeHostWrapper lets a host object as parameter and get back a host object from return value, this is useful when declare custom transport layer and use it with this {@link MessageBusConnection}
     */
    public constructor(private bridgeHostCreator: () => Promise<IRpcBridgeHost<any>>, private bridgeHostWrapper: (wrappedBridgeHostCreator: () => Promise<IRpcBridgeHost<any>>) => Promise<IRpcBridgeHost<any>> = async wrappedBridgeHostCreator => await wrappedBridgeHostCreator()) {}
    /**
     * Connect to MessageBus server
     */
    public async connect() {
        this.bridge = new RpcBridge<IMessageBusClientInterface, IMessageBusServerInterface>(
            await this.bridgeHostWrapper(async () => new KeepAliveRpcBridgeHost(await this.bridgeHostCreator())),
            new MessageBusClientImplementation(this)
        )
        this.bridge.onCommunicationClosed = this._onUnderlyingConnectionIsClosed.bind(this)
        this.bridge.onCommunicationRecovered = this._onUnderlyingConnectionIsRecovered.bind(this)
        await this.bridge.connect()
    }
    /**
     * Close all server hosts when underlying message bus connection is lost.
     * @private
     */
    private async _onUnderlyingConnectionIsClosed() {
        // const promises: Promise<void>[] = []
        this._hosts.forEach((host, key) => {
            // promises.push(host.shutdown())
            host.shutdownOnlyBackend()
        })
        // await Promise.all(promises)
    }
    private async _onUnderlyingConnectionIsRecovered() {
        const promises: Promise<void>[] = []
        this._hosts.forEach((host, key) => {
            promises.push(host.attemptToReCreateUnderlyingServerHost())
        })
        await Promise.all(promises)
    }
    /**
     * Create a new {@link IRpcBridgeHost} instance over this MessageBus connection.
     * @remarks
     * {@link localVirtualHost} is represent to client socket ip address on socket programming also {@link remoteVirtualHost} is represent to server socket ip address too.
     * @param localVirtualHost for example: "MyMessageBusServer::MyClient1" or "xx.yy.zz.ww/xxClient" etc, this is just (and must) be unique.
     * @param remoteVirtualHost for example: "MyMessageBusServer" or "xx.yy.zz.ww" etc, this is the server virtual host name you want to connect to.
     * @returns new {@link IRpcBridgeHost} instance over this message bus connection
     */
    public async createBridgeHost(remoteVirtualHost: string, localVirtualHost: string): Promise<IRpcBridgeHost<any>> {
        const host = new MessageBusRpcBridgeHost(this, this.bridge.localInterface as MessageBusClientImplementation, { local: new VirtualChannel(`${remoteVirtualHost}/Clients/${localVirtualHost}`), remote: new VirtualChannel(remoteVirtualHost) })
        return host
    }
    /**
     * @param virtualHost Virtual host name, this is must be unique and can be anything, this is correspond Domain Name on http servers
     * @param attemptToRecreateConnectionAfterMilliSeconds When underlying connection is lost, {@link MessageBusConnection} is attempt to reconnect to message bus and recreate this server host in internally after this time.
     * @example rpc://MyServer.com or xxx.yyy.zzz and any unique string.
     * @returns a new {@link IRpcServerHost} instance over this message bus connection
     */
    public async createServerHost(virtualHost: string, attemptToRecreateConnectionAfterMilliSeconds = 1000 * 60): Promise<IRpcServerHost<any>> {
        const id = this._allocateId()
        const host = new IntermediateServerHost(id, async () => new MessageBusRpcServerHost(this.bridge.localInterface as MessageBusClientImplementation, this, new VirtualChannel(virtualHost)), attemptToRecreateConnectionAfterMilliSeconds)
        host.onShutdown = async () => {
            this._hosts.delete(id)
        }
        this._hosts.set(id, host)
        return host
    }
}
/**
 * Used by internally only, users not need to use this
 * used to auto recover server hosts over {@link MessageBusConnection} when {@link MessageBusConnection} underlying connection is recovered.
 */
class IntermediateServerHost implements IRpcServerHost<IRpcServerHost<any>> {
    backend: IRpcServerHost<any> = null!
    private handler: ServerHostEvents = null!
    private _timerId: NodeJS.Timeout = null!
    public onError = async (error: Error) => {
        console.error(error)
    }
    public onShutdown = async () => {}

    public constructor(public id: number, private hostCreator: () => Promise<IRpcServerHost<any>>, private attemptToRecreateConnectionAfterMilliSeconds: number) {
    }

    public async launch(): Promise<void> {
        await this.backend.launch()
    }

    public async prepare(eventHandler: ServerHostEvents): Promise<void> {
        this.handler = new ServerHostEvents(
            eventHandler.onReady,
            eventHandler.onClientConnected,
            eventHandler.onError,
            async () => {
                await eventHandler.onShutdown()
            }
        )
        this.backend = await this.hostCreator()
        await this.backend.prepare(this.handler)
    }

    public async attemptToReCreateUnderlyingServerHost() {
        this._timerId = setInterval(this._onPeriodic.bind(this), this.attemptToRecreateConnectionAfterMilliSeconds)
    }
    private async _onPeriodic() {
        try {
            this.backend = await this.hostCreator()
            await this.prepare(this.handler)
            await this.launch()
            clearInterval(this._timerId)
        } catch (error) {
            await this.onError(error)
        }
    }
    public async shutdownOnlyBackend() {
        await this.backend.shutdown()
    }
    public async shutdown(): Promise<void> {
        clearInterval(this._timerId)
        await this.onShutdown()
        await this.shutdownOnlyBackend()
    }
}
